package io.github.wanggit.antrpc;

import io.github.wanggit.antrpc.client.config.IConfigContainer;
import io.github.wanggit.antrpc.client.config.listeners.DefaultConfigChangeListener;
import io.github.wanggit.antrpc.client.config.listeners.Slf4LoggerLevelChangeListener;
import io.github.wanggit.antrpc.client.monitor.IRpcCallLogHolder;
import io.github.wanggit.antrpc.client.monitor.RpcCallLogHolder;
import io.github.wanggit.antrpc.client.spring.*;
import io.github.wanggit.antrpc.client.zk.IZkClient;
import io.github.wanggit.antrpc.client.zk.ZkClient;
import io.github.wanggit.antrpc.client.zk.lb.ILoadBalancerContainer;
import io.github.wanggit.antrpc.client.zk.lb.LoadBalancerContainer;
import io.github.wanggit.antrpc.client.zk.lb.LoadBalancerHelper;
import io.github.wanggit.antrpc.client.zk.listener.*;
import io.github.wanggit.antrpc.client.zk.register.IRegister;
import io.github.wanggit.antrpc.client.zk.register.IZkRegisterHolder;
import io.github.wanggit.antrpc.client.zk.register.ZkRegister;
import io.github.wanggit.antrpc.client.zk.register.ZkRegisterHolder;
import io.github.wanggit.antrpc.client.zk.zknode.*;
import io.github.wanggit.antrpc.commons.breaker.IResilience4jCircuitBreakerContainer;
import io.github.wanggit.antrpc.commons.breaker.Resilience4jCircuitBreakerContainer;
import io.github.wanggit.antrpc.commons.codec.cryption.CodecHolder;
import io.github.wanggit.antrpc.commons.codec.cryption.ICodecHolder;
import io.github.wanggit.antrpc.commons.codec.serialize.ISerializerHolder;
import io.github.wanggit.antrpc.commons.codec.serialize.SerializerHolder;
import io.github.wanggit.antrpc.commons.config.Configuration;
import io.github.wanggit.antrpc.commons.config.IConfiguration;
import io.github.wanggit.antrpc.commons.config.TelnetConfig;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.rate.IResilience4jRateLimiter;
import io.github.wanggit.antrpc.commons.rate.Resilience4jRateLimiterContainer;
import io.github.wanggit.antrpc.commons.utils.NetUtil;
import io.github.wanggit.antrpc.server.IServer;
import io.github.wanggit.antrpc.server.RpcServer;
import io.github.wanggit.antrpc.server.invoker.IRpcRequestBeanInvoker;
import io.github.wanggit.antrpc.server.invoker.RpcRequestBeanInvoker;
import io.github.wanggit.antrpc.server.telnet.ITelnetServer;
import io.github.wanggit.antrpc.server.telnet.TelnetServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;

import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class AntrpcContext implements IAntrpcContext {

    private static final String ANTRPC_CONTEXT_BEAN_NAME = "antrpcContext";

    private BeanContainer beanContainer;

    private final IConfiguration configuration;

    private IResilience4jCircuitBreakerContainer resilience4jCircuitBreakerContainer;

    private IResilience4jRateLimiter clientResilience4jRateLimiterContainer;

    private IResilience4jRateLimiter serverResilience4jRateLimiterContainer;

    private IRpcCallLogHolder rpcCallLogHolder;

    private IZkClient zkClient;

    private IZkRegisterHolder zkRegisterHolder;

    private LoadBalancerHelper loadBalancerHelper;

    private INodeHostContainer nodeHostContainer;

    private IRpcRequestBeanInvoker rpcRequestBeanInvoker;

    private final AtomicBoolean inited = new AtomicBoolean(false);

    private IRegister register;

    private IServer server;

    private ITelnetServer telnetServer;

    private IRpcClients rpcClients;

    private IOnFailHolder onFailHolder;

    private ICodecHolder codecHolder;

    private ISerializerHolder serializerHolder;

    private IListener lbNodeChangeListener;

    private IListener registerNodeChangeListener;

    private IListener subscribeNodeChangeListener;

    private IListener unregNodeChangeListener;

    private IZkNodeCleaner zkNodeCleaner;

    private IOnFailProcessor onFailProcessor;

    private IRpcAutowiredProcessor rpcAutowiredProcessor;

    private CallerProxyFactory callerProxyFactory;

    private CallerProxyMethodInterceptor callerProxyMethodInterceptor;

    private IReportSubscriber reportSubscriber;

    private IZkNodeOperator zkNodeOperator;

    private ILoadBalancerContainer loadBalancerContainer;

    private ConfigurableApplicationContext springApplicationContext;

    public AntrpcContext(IConfiguration configuration) {
        this.configuration = configuration;
    }

    @Override
    public ConfigurableApplicationContext getSpringApplicationContext() {
        return springApplicationContext;
    }

    @Override
    public boolean isInited() {
        return inited.get();
    }

    @Override
    public CallerProxyFactory getCallerProxyFactory() {
        return callerProxyFactory;
    }

    @Override
    public CallerProxyMethodInterceptor getCallerProxyMethodInterceptor() {
        return callerProxyMethodInterceptor;
    }

    @Override
    public IZkNodeCleaner getZkNodeCleaner() {
        return zkNodeCleaner;
    }

    @Override
    public IZkNodeOperator getZkNodeOperator() {
        return zkNodeOperator;
    }

    @Override
    public ISerializerHolder getSerializerHolder() {
        return serializerHolder;
    }

    @Override
    public ICodecHolder getCodecHolder() {
        return codecHolder;
    }

    @Override
    public IOnFailHolder getOnFailHolder() {
        return onFailHolder;
    }

    @Override
    public IResilience4jRateLimiter getClientResilience4jRateLimiterContainer() {
        return clientResilience4jRateLimiterContainer;
    }

    @Override
    public IResilience4jRateLimiter getServerResilience4jRateLimiterContainer() {
        return serverResilience4jRateLimiterContainer;
    }

    @Override
    public IRpcClients getRpcClients() {
        return rpcClients;
    }

    @Override
    public BeanContainer getBeanContainer() {
        return beanContainer;
    }

    @Override
    public IConfiguration getConfiguration() {
        return configuration;
    }

    @Override
    public IResilience4jCircuitBreakerContainer getResilience4jCircuitBreakerContainer() {
        return resilience4jCircuitBreakerContainer;
    }

    @Override
    public IRpcCallLogHolder getRpcCallLogHolder() {
        return rpcCallLogHolder;
    }

    @Override
    public IZkClient getZkClient() {
        return zkClient;
    }

    @Override
    public IZkRegisterHolder getZkRegisterHolder() {
        return zkRegisterHolder;
    }

    @Override
    public IRegister getRegister() {
        return register;
    }

    @Override
    public ILoadBalancerContainer getLoadBalancerContainer() {
        return loadBalancerContainer;
    }

    public void setRegister(IRegister register) {
        this.register = register;
    }

    public void setOnFailProcessor(IOnFailProcessor onFailProcessor) {
        this.onFailProcessor = onFailProcessor;
    }

    public void setRpcAutowiredProcessor(IRpcAutowiredProcessor rpcAutowiredProcessor) {
        this.rpcAutowiredProcessor = rpcAutowiredProcessor;
    }

    @Override
    public IRpcAutowiredProcessor getRpcAutowiredProcessor() {
        return rpcAutowiredProcessor;
    }

    @Override
    public IOnFailProcessor getOnFailProcessor() {
        return onFailProcessor;
    }

    @Override
    public IListener getLbNodeChangeListener() {
        return lbNodeChangeListener;
    }

    @Override
    public IListener getRegisterNodeChangeListener() {
        return registerNodeChangeListener;
    }

    @Override
    public IListener getSubscribeNodeChangeListener() {
        return subscribeNodeChangeListener;
    }

    @Override
    public IListener getUnregNodeChangeListener() {
        return unregNodeChangeListener;
    }

    @Override
    public LoadBalancerHelper getLoadBalancerHelper() {
        return loadBalancerHelper;
    }

    @Override
    public INodeHostContainer getNodeHostContainer() {
        return nodeHostContainer;
    }

    @Override
    public IRpcRequestBeanInvoker getRpcRequestBeanInvoker() {
        return rpcRequestBeanInvoker;
    }

    @Override
    public void init(ConfigurableApplicationContext applicationContext) {
        // 不重复初始化
        if (!inited.compareAndSet(false, true)) {
            return;
        }
        long start = System.currentTimeMillis();
        springApplicationContext = applicationContext;
        doAntRpcBeanAnnotationCheck(applicationContext.getBeanFactory());
        doRegisterAntrpcContextToSpring(applicationContext.getBeanFactory());
        initExposedIp(configuration);
        initRpcCallLogHolder(applicationContext, configuration);
        // 服务提供方频控限流
        serverResilience4jRateLimiterContainer = new Resilience4jRateLimiterContainer();
        initRpcRequestBeanInvoker(applicationContext, serverResilience4jRateLimiterContainer);
        // 创建服务调用方调用失败时的默认处理器Holder
        onFailHolder = new OnFailHolder();
        initOnFailProcessor(onFailHolder);
        initZkClient(configuration);
        initZkNodeOperator(zkClient);
        initLoadBalancerHelper(configuration);
        // 服务调用方频控限流
        clientResilience4jRateLimiterContainer = new Resilience4jRateLimiterContainer();
        initLoadBalancerContainer(loadBalancerHelper, clientResilience4jRateLimiterContainer);
        initNodeHostContainer(configuration, loadBalancerContainer, zkNodeOperator);
        initZkNodeCleaner(zkNodeOperator);
        initZkRegisterHolder(register, zkNodeOperator, configuration);
        initCircuitBreaker(configuration);
        initCodecHolder(configuration);
        initSerializerHolder(configuration);
        // RPC 客户端容器
        rpcClients = new RpcClients(configuration, codecHolder, serializerHolder);
        initRpcBeanContainer(
                rpcCallLogHolder,
                onFailHolder,
                resilience4jCircuitBreakerContainer,
                clientResilience4jRateLimiterContainer,
                rpcClients,
                serializerHolder,
                nodeHostContainer);
        initReportSubscriber(zkNodeOperator, nodeHostContainer);
        initRpcAutowiredProcessor(beanContainer, reportSubscriber, configuration);
        initAllListeners(
                zkClient,
                zkRegisterHolder,
                nodeHostContainer,
                configuration,
                serverResilience4jRateLimiterContainer,
                loadBalancerContainer,
                reportSubscriber);
        beginRegister(zkNodeOperator, zkRegisterHolder, configuration);
        startServer(configuration, codecHolder, rpcRequestBeanInvoker, serializerHolder);
        startTelnetServer(configuration, applicationContext);
        initShutdownHook(applicationContext);
        // 初始化远程配置变更监听器
        initDefaultConfigChangeListener(configuration);
        if (log.isInfoEnabled()) {
            log.info(ConstantValues.BANNER);
            log.info(
                    "Antrpc startup completed, "
                            + (System.currentTimeMillis() - start)
                            + "ms was used.");
        }
    }

    private void initDefaultConfigChangeListener(IConfiguration configuration) {
        try {
            springApplicationContext
                    .getBean(IConfigContainer.class)
                    .addConfigChangeListener(
                            new DefaultConfigChangeListener(configuration, this),
                            new Slf4LoggerLevelChangeListener());
        } catch (NoSuchBeanDefinitionException e) {
            if (log.isWarnEnabled()) {
                log.warn("Not Found " + IConfigContainer.class.getName());
            }
        }
    }

    /**
     * 注册JVM关闭时的钓子，用于向Zookeeper中注销节点。
     *
     * @param applicationContext spring context
     */
    private void initShutdownHook(ConfigurableApplicationContext applicationContext) {
        Runtime.getRuntime()
                .addShutdownHook(
                        new Thread(
                                () -> applicationContext.getBean(IAntrpcContext.class).destroy(),
                                "AntRpcShutdownHook"));
    }

    private void initZkNodeOperator(IZkClient zkClient) {
        zkNodeOperator = new ZkNodeOperator(zkClient.getCurator());
    }

    /**
     * 创建负载均衡容器，其中包含客户端限流频控
     *
     * @param loadBalancerHelper helper
     * @param clientResilience4jRateLimiter client rate limiter
     */
    private void initLoadBalancerContainer(
            LoadBalancerHelper loadBalancerHelper,
            IResilience4jRateLimiter clientResilience4jRateLimiter) {
        loadBalancerContainer =
                new LoadBalancerContainer(loadBalancerHelper, clientResilience4jRateLimiter);
    }

    private void initReportSubscriber(
            IZkNodeOperator zkNodeOperator, INodeHostContainer nodeHostContainer) {
        reportSubscriber = new ReportSubscribe(zkNodeOperator, nodeHostContainer);
    }

    @Override
    public void destroy() {
        if (log.isInfoEnabled()) {
            log.info("AntrpcContext is being destroyed.");
        }
        // 注销所有接口
        register.unregister(configuration, zkNodeOperator);
        // 删除订阅接口列表
        reportSubscriber.unReport();
        zkNodeCleaner.destory();
        zkRegisterHolder.destroy();
        if (null != server) {
            server.close();
        }
        if (null != telnetServer) {
            telnetServer.close();
        }
    }

    /**
     * 为对外发起RPC调用的对象创建代理对象。
     *
     * @param beanFactory bean factory
     */
    private void doAntRpcBeanAnnotationCheck(ConfigurableListableBeanFactory beanFactory) {
        register = new ZkRegister();
        onFailProcessor = new OnFailProcessor();
        rpcAutowiredProcessor = new RpcAutowiredProcessor();
        callerProxyFactory = new DefaultCallerProxyFactory();
        callerProxyMethodInterceptor = new CallerProxyMethodInterceptor();
        String[] names = beanFactory.getBeanDefinitionNames();
        for (String name : names) {
            Object bean = internalGetBean(beanFactory, name);
            if (null != bean) {
                // 为Bean创建代理对象，处理上下游的依赖关系
                bean =
                        callerProxyFactory.proxy(
                                name, bean, beanFactory, callerProxyMethodInterceptor);
                // 处理有属性被@RpcAutowired标记的对象
                rpcAutowiredProcessor.checkBeanHasRpcAutowire(bean);
                // 处理被@OnRpcFail标记的用于失败处理回调的对象
                onFailProcessor.checkHasOnRpcFail(bean);
                // 处理被@RpcService标记的对象，并准备发布为RPC服务
                register.checkHasRpcService(bean);
            }
        }
    }

    /**
     * 把当前AntRpcContext上下文以单例的方式添加到Spring容器中
     *
     * @param beanFactory spring bean factory
     */
    private void doRegisterAntrpcContextToSpring(ConfigurableListableBeanFactory beanFactory) {
        if (!beanFactory.containsBean(ANTRPC_CONTEXT_BEAN_NAME)) {
            beanFactory.registerSingleton(ANTRPC_CONTEXT_BEAN_NAME, this);
        }
    }

    private Object internalGetBean(ConfigurableListableBeanFactory beanFactory, String name) {
        try {
            return beanFactory.getBean(name);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("This Bean cannot be found in the Spring container. name = " + name, e);
            }
            return null;
        }
    }

    /**
     * 服务器有多张网卡时，可选择在Zookeeper中暴露那张网卡的IP地址，默认情况下取本机IP
     *
     * @param configuration cfg
     */
    private void initExposedIp(IConfiguration configuration) {
        if (null == configuration.getExposeIp()) {
            ((Configuration) configuration).setExposeIp(NetUtil.getLocalIp());
        }
    }

    private void initRpcAutowiredProcessor(
            BeanContainer beanContainer,
            IReportSubscriber reportSubscriber,
            IConfiguration configuration) {
        if (null == beanContainer) {
            throw new IllegalArgumentException();
        }
        rpcAutowiredProcessor.init(beanContainer, reportSubscriber, configuration);
    }

    /**
     * 把 io.github.wanggit.antrpc.client.spring.IOnFailProcessor#checkHasOnRpcFail 方法收集的@OnRpcFail
     * 对象 bean按 io.github.wanggit.antrpc.commons.annotations.OnRpcFail#clazz 为键，bean为值 整理到Map中
     *
     * @param onFailHolder holder
     */
    private void initOnFailProcessor(IOnFailHolder onFailHolder) {
        if (null == onFailHolder) {
            throw new IllegalArgumentException();
        }
        onFailProcessor.init(onFailHolder);
    }

    /**
     * 初始化Zookeeper中的过期节点清理工具
     *
     * @param zkNodeOperator zkOpr
     */
    private void initZkNodeCleaner(IZkNodeOperator zkNodeOperator) {
        if (null == zkNodeOperator) {
            throw new IllegalArgumentException();
        }
        zkNodeCleaner = new ZkNodeCleaner(zkNodeOperator);
    }

    /**
     * 初始化所有Listener
     *
     * @param zkClient zk client
     * @param zkRegisterHolder register holder
     * @param nodeHostContainer node container
     * @param configuration cfg
     * @param serverResilience4jRateLimiterContainer server rate limiter container
     * @param loadBalancerContainer lb container
     * @param reportSubscriber report subscriber
     */
    private void initAllListeners(
            IZkClient zkClient,
            IZkRegisterHolder zkRegisterHolder,
            INodeHostContainer nodeHostContainer,
            IConfiguration configuration,
            IResilience4jRateLimiter serverResilience4jRateLimiterContainer,
            ILoadBalancerContainer loadBalancerContainer,
            IReportSubscriber reportSubscriber) {
        // Zookeeper 中的注册服务节点变化监听器，"/__rpc_sub__/" 节点
        subscribeNodeChangeListener =
                new SubscribeNodeChangeListener(zkClient, loadBalancerContainer, configuration);
        // 监听Zookeeper中节点注销的监听器 "/__rpc_unreg__/" 节点
        unregNodeChangeListener = new UnregNodeChangeListener(zkClient, zkRegisterHolder);
        // 监听Zookeeper中负载均衡变化的监听器 "/__rpc_lbs__/" 节点
        lbNodeChangeListener =
                new LbNodeChangeListener(zkClient, loadBalancerContainer, reportSubscriber);
        // 监听Zookeeper中新注册的服务名录 "/__rpc_na__/" 节点
        registerNodeChangeListener =
                new RegisterNodeChangeListener(
                        zkClient,
                        zkRegisterHolder,
                        nodeHostContainer,
                        configuration,
                        serverResilience4jRateLimiterContainer,
                        reportSubscriber);
        subscribeNodeChangeListener.listen();
        unregNodeChangeListener.listen();
        lbNodeChangeListener.listen();
        registerNodeChangeListener.listen();
    }

    /**
     * 创建服务提供者端的请求执行器，用于执行收到的服务调用方的请求。
     *
     * @param applicationContext application context
     * @param serverResilience4jRateLimiter server rate limiter
     */
    private void initRpcRequestBeanInvoker(
            ConfigurableApplicationContext applicationContext,
            IResilience4jRateLimiter serverResilience4jRateLimiter) {
        rpcRequestBeanInvoker =
                new RpcRequestBeanInvoker(
                        applicationContext.getBeanFactory(), serverResilience4jRateLimiter);
    }

    /**
     * 开始注册本服务对外提供的RPC服务
     *
     * @param zkNodeOperator zk opr
     * @param zkRegisterHolder register holder
     * @param configuration cfg
     */
    private void beginRegister(
            IZkNodeOperator zkNodeOperator,
            IZkRegisterHolder zkRegisterHolder,
            IConfiguration configuration) {
        if (null == zkNodeOperator || null == zkRegisterHolder || null == configuration) {
            throw new IllegalArgumentException();
        }
        register.init(zkNodeOperator, zkRegisterHolder, configuration);
    }

    /**
     * 创建服务调用方的RPC接口代理类容器
     *
     * @param rpcCallLogHolder log holder
     * @param onFailHolder on fail holder
     * @param resilience4jCircuitBreakerContainer circuit breaker container
     * @param clientResilience4jRateLimiterContainer client rate limiter container
     * @param rpcClients rpc clients
     * @param serializerHolder serializer holder
     * @param nodeHostContainer node container
     */
    private void initRpcBeanContainer(
            IRpcCallLogHolder rpcCallLogHolder,
            IOnFailHolder onFailHolder,
            IResilience4jCircuitBreakerContainer resilience4jCircuitBreakerContainer,
            IResilience4jRateLimiter clientResilience4jRateLimiterContainer,
            IRpcClients rpcClients,
            ISerializerHolder serializerHolder,
            INodeHostContainer nodeHostContainer) {
        beanContainer =
                new RpcBeanContainer(
                        rpcCallLogHolder,
                        onFailHolder,
                        resilience4jCircuitBreakerContainer,
                        clientResilience4jRateLimiterContainer,
                        rpcClients,
                        serializerHolder,
                        nodeHostContainer);
    }

    /**
     * 初始化熔断器配置容器，包含全局熔断器与接口熔断器
     *
     * @param configuration cfg
     */
    private void initCircuitBreaker(IConfiguration configuration) {
        if (null == configuration) {
            throw new IllegalArgumentException();
        }
        resilience4jCircuitBreakerContainer =
                new Resilience4jCircuitBreakerContainer(configuration);
    }

    /**
     * 初始化序列化工具容器
     *
     * @param configuration cfg
     */
    private void initSerializerHolder(IConfiguration configuration) {
        if (null == configuration) {
            throw new IllegalArgumentException();
        }
        try {
            serializerHolder = new SerializerHolder(configuration);
        } catch (Exception e) {
            throw new BeanCreationException(
                    "SerializerHolder initialization failed, please heartBeatWasContinuousLoss that the configured [antrpc.serialize-type = "
                            + configuration.getSerializeConfig().getType()
                            + "] was correct.");
        }
    }

    /**
     * 初始化消息加解密器
     *
     * @param configuration cfg
     */
    private void initCodecHolder(IConfiguration configuration) {
        if (null == configuration) {
            throw new IllegalArgumentException();
        }
        try {
            codecHolder = new CodecHolder(configuration.getCodecConfig());
        } catch (Exception e) {
            throw new BeanCreationException(
                    "An exception occurred while initializing " + CodecHolder.class.getName(), e);
        }
    }

    private void initZkRegisterHolder(
            IRegister register, IZkNodeOperator zkNodeOperator, IConfiguration configuration) {
        if (null == register || null == zkNodeOperator || null == configuration) {
            throw new IllegalArgumentException();
        }
        zkRegisterHolder = new ZkRegisterHolder(register, zkNodeOperator, configuration);
    }

    /**
     * 初始化服务节点容器
     *
     * @param configuration cfg
     * @param loadBalancerContainer lb container
     * @param zkNodeOperator zk node opr
     */
    private void initNodeHostContainer(
            IConfiguration configuration,
            ILoadBalancerContainer loadBalancerContainer,
            IZkNodeOperator zkNodeOperator) {
        if (null == configuration || null == loadBalancerContainer) {
            throw new IllegalArgumentException();
        }
        nodeHostContainer =
                new NodeHostContainer(
                        loadBalancerContainer, configuration.getDirectHosts(), zkNodeOperator);
    }

    private void initLoadBalancerHelper(IConfiguration configuration) {
        if (null == configuration) {
            throw new IllegalArgumentException();
        }
        loadBalancerHelper = new LoadBalancerHelper(configuration);
    }

    /**
     * 初始化Zookeeper的客户端
     *
     * @param configuration cfg
     */
    private void initZkClient(IConfiguration configuration) {
        if (null == configuration) {
            throw new IllegalArgumentException();
        }
        zkClient = new ZkClient(configuration);
    }

    /**
     * 初始化RPC调用的日志上报工具
     *
     * @param applicationContext application context
     * @param configuration cfg
     */
    private void initRpcCallLogHolder(
            ApplicationContext applicationContext, IConfiguration configuration) {
        if (null == configuration) {
            throw new IllegalArgumentException();
        }
        try {
            rpcCallLogHolder = new RpcCallLogHolder(configuration, applicationContext);
        } catch (Exception e) {
            throw new BeanCreationException(
                    "An exception occurred while initializing " + IRpcCallLogHolder.class.getName(),
                    e);
        }
    }

    public IServer getServer() {
        return server;
    }

    /**
     * 启动RPC服务端
     *
     * @param configuration cfg
     * @param codecHolder codec holder
     * @param rpcRequestBeanInvoker invoker
     * @param serializerHolder serializer holder
     */
    private void startServer(
            IConfiguration configuration,
            ICodecHolder codecHolder,
            IRpcRequestBeanInvoker rpcRequestBeanInvoker,
            ISerializerHolder serializerHolder) {
        if (null == configuration
                || null == codecHolder
                || null == rpcRequestBeanInvoker
                || null == serializerHolder) {
            throw new IllegalArgumentException();
        }
        if (configuration.isStartServer()) {
            server =
                    new RpcServer(
                            configuration, codecHolder, rpcRequestBeanInvoker, serializerHolder);
            try {
                server.open(configuration.getPort());
            } catch (InterruptedException e) {
                if (log.isErrorEnabled()) {
                    log.error("Antrpc server failed to start.", e);
                }
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 启用Telnet运维命令服务器
     *
     * @param configuration cfg
     * @param applicationContext spring context
     */
    private void startTelnetServer(
            IConfiguration configuration, ConfigurableApplicationContext applicationContext) {
        TelnetConfig telnetConfig = configuration.getTelnetConfig();
        if (null != telnetConfig.getEnable() && telnetConfig.getEnable()) {
            telnetServer = new TelnetServer(this, applicationContext);
            try {
                telnetServer.start(telnetConfig);
            } catch (Exception e) {
                if (log.isErrorEnabled()) {
                    log.error("TelnetServer failed to start.", e);
                }
            }
        }
    }

    @Override
    public synchronized void restartTelnetServerIfRunning(
            IConfiguration configuration, ConfigurableApplicationContext applicationContext) {
        if (null != telnetServer && telnetServer.isRunning()) {
            telnetServer.close();
            telnetServer = null;
        }
        startTelnetServer(configuration, applicationContext);
    }

    @Override
    public synchronized void stopTelnetServerIfRunning(
            IConfiguration configuration, ConfigurableApplicationContext springApplicationContext) {
        if (null != telnetServer && telnetServer.isRunning()) {
            telnetServer.close();
            telnetServer = null;
        }
    }
}
